#pragma once
#include <string>
#include <iostream>

#include <memory>
#include <functional>
#include <mutex>
#include <unordered_map>

#include "route.hpp"
#include "virtualhost.hpp"
#include "consumer.hpp"
#include "../mqcommom/threadpool.hpp"
#include "../mqcommom/helper.hpp"
#include "../mqcommom/proto.pb.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"

namespace ns_rabbitmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using OpenChannelRequestptr = std::shared_ptr<OpenChannelRequest>;
    using CloseChannelRequestptr = std::shared_ptr<CloseChannelRequest>;
    using DeclearExchangeRequestptr = std::shared_ptr<DeclearExchangeRequest>;
    using DeleteExchangeRequestptr = std::shared_ptr<DeleteExchangeRequest>;
    using DeclearQueueRequestptr = std::shared_ptr<DeclearQueueRequest>;
    using DeleteQueueRequestptr = std::shared_ptr<DeleteQueueRequest>;
    using QueueBindRequestptr = std::shared_ptr<QueueBindRequest>;
    using QueueUnbindRequestptr = std::shared_ptr<QueueUnbindRequest>;
    using BasicPublishRequestptr = std::shared_ptr<BasicPublishRequest>;
    using BasicConsumerRequestptr = std::shared_ptr<BasicConsumerRequest>;
    using BasicAckRequestptr = std::shared_ptr<BasicAckRequest>;
    using BasicCancelRequestptr = std::shared_ptr<BasicCancelRequest>;
    using BasicConsumerResponseptr = std::shared_ptr<BasicConsumerResponse>;
    using BasicCommResponseptr = std::shared_ptr<BasicCommResponse>;

    class Channel
    {
    public:
        Channel(const std::string &cid, const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const ConsumerManangerPtr &cmp, const HostPtr &hostptr, const ThreadPoolPtr &poolptr)
            : _cid(cid), _conn(conn), _codec(codec), _cmp(cmp), _hostptr(hostptr), _poolptr(poolptr)
        {
        }

        // 信道关闭的时候，要将消费者销毁
        ~Channel()
        {
            if (_consumer.get())
            {
                _cmp->Remove(_consumer->_tag, _consumer->_qname);
            }
            DLOG("del Channel: %s",_cid.c_str() );
        }

        void DeclearExchange(const DeclearExchangeRequestptr &req)
        {
            bool ret = _hostptr->DeclareExchange(req->exchange_name(), req->exchange_type(), req->durable(),
                                                 req->auto_delect(), req->args());
            return BasicResponse(ret, req->rid(), req->cid());
        }

        void DeleteExchange(const DeleteExchangeRequestptr &req)
        {
            _hostptr->DeleteExchange(req->exchange_name());
            return BasicResponse(true, req->rid(), req->cid());
        }

        void DeclareQueue(const DeclearQueueRequestptr &req)
        {
            bool ret = _hostptr->DeclareQueue(req->queue_name(), req->exclusive(), req->durable(), req->auto_delect(), req->args());
            _cmp->InitQueueConsumer(req->queue_name());
            return BasicResponse(ret, req->rid(), req->cid());
        }
        void DeleteQueue(const DeleteQueueRequestptr &req)
        {
            _hostptr->DeleteQueue(req->queue_name());
            return BasicResponse(true, req->rid(), req->cid());
        }

        void QueueBind(const QueueBindRequestptr &req)
        {
            bool ret = _hostptr->Bind(req->exchange_name(), req->queue_name(), req->bindkey(), req->durable());
            return BasicResponse(ret, req->rid(), req->cid());
        }
        void QueueUnbind(const QueueUnbindRequestptr &req)
        {
            _hostptr->UnBind(req->exchange_name(), req->queue_name());
            return BasicResponse(true, req->rid(), req->cid());
        }

        void BasicPublish(const BasicPublishRequestptr &req)
        {
            // 1.判断交换机是否存在
            auto ep = _hostptr->SelectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                DLOG("不存在指定的交换机%s", req->exchange_name().c_str());
                return BasicResponse(false, req->rid(), req->cid());
            }
            BasicProperties *properties = nullptr;
            std::string routingkey;
            if (req->has_propertie())
            {
                properties = req->mutable_propertie();
                routingkey = properties->routing_key();
            }
            // 2.获取所有的关联队列
            QueueMap qmps = _hostptr->GetExchangeBindings(req->exchange_name());
            for (auto &que : qmps)
            {
                std::string bindingkey = que.second->_binding_key;
                Router::IsBindingKeylegal(bindingkey);
                Router::IsRoutingKeylegal(routingkey);
                // 3.判断是否路由
                if (Router::Route(ep->_exchange_type, routingkey, bindingkey))
                {
                    DLOG("rotingkey:%s - bindingkey:%s 匹配成功_发送的消息是:%s", routingkey.c_str(), bindingkey.c_str(), req->body().c_str());
                    // 4.将消息添加到队列中
                    _hostptr->BasicPublish(que.first, properties, req->body());
                    // 5.添加到线程中
                    auto task = std::bind(&Channel::Consume, this, que.first);
                    _poolptr->push(task);
                }
            }
            return BasicResponse(true, req->rid(), req->cid());
        }

        void Consume(std::string &qname)
        {
            // 1.取出一个任务
            MessagePtr msgptr = _hostptr->BasicConsume(qname);
            if (!msgptr.get())
            {
                DLOG("任务不存在!");
                return;
            }
            // 2.取出一个消费者
            ConsumerPtr csmptr = _cmp->Chose(qname);
            if (!csmptr.get())
            {
                DLOG("队列%s的消费者不存在", qname.c_str());
                return;
            }
            ILOG("取出队列%s:%s进行消费任务", qname.c_str(), csmptr->_tag.c_str());
            // 3.调用消费者的回调处理任务
            csmptr->_cb(csmptr->_tag, msgptr->mutable_payload()->mutable_basic_properties(), msgptr->payload().body());
            if (csmptr->_auto_delete)
            {
                _hostptr->Ack(qname, msgptr->payload().basic_properties().id());
            }
        }

        // 订阅队列
        void BasicConsume(const BasicConsumerRequestptr &req)
        {
            // 1.检查队列是否存在
            bool ret = _hostptr->IsQueueExists(req->qname());
            if (!ret)
            {
                DLOG("不存在当前队列%s,无法订阅", req->qname().c_str());
                return BasicResponse(false, req->rid(), req->cid());
            }
            // 2.创建消费者
            auto cb = std::bind(&Channel::CallBack, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);

            _consumer = _cmp->Create(req->consumer_tag(), req->qname(), req->auto_ack(), cb);
            DLOG("创建了消费:%s", req->consumer_tag().c_str());
            // 3，调用基本response
            return BasicResponse(true, req->rid(), req->cid());
        }

        // 取消订阅
        void BasicCancel(const BasicCancelRequestptr &req)
        {
            _cmp->Remove(req->consumer_tag(), req->qname());
            return BasicResponse(true, req->rid(), req->cid());
        }
        void BasicAck(const BasicAckRequestptr &req)
        {
            _hostptr->Ack(req->queue_name(), req->msg_id());
            return BasicResponse(true, req->rid(), req->cid());
        }

    private:
        void BasicResponse(bool ok, const std::string &rid, const std ::string &cid)
        {
            BasicCommResponseptr resp = std::make_shared<BasicCommResponse>();
            resp->set_rid(rid);
            resp->set_cid(cid);
            resp->set_ok(ok);
            _codec->send(_conn, *resp);
        }

        // 组织消息，往对应信道上的客户端发送消息
        void CallBack(const std::string &tag, BasicProperties *bp, const std::string &body)
        {
            BasicConsumerResponse resp;
            resp.set_body(body);
            resp.set_cid(_cid);
            resp.set_consumer_tag(tag);
            if (bp)
            {
                resp.mutable_propertie()->set_id(bp->id());
                resp.mutable_propertie()->set_deliver_mode(bp->deliver_mode());
                resp.mutable_propertie()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }

    private:
        std::string _cid; // 信道标志
        ConsumerPtr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ConsumerManangerPtr _cmp; // 消费者队列管理句柄
        HostPtr _hostptr;
        ThreadPoolPtr _poolptr;
    };

    using ChannelPtr = std::shared_ptr<Channel>;
    class ChannelManager
    {
    public:
        ChannelManager() {}
        bool OpenChannel(const std::string &cid, const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec,
                         const ConsumerManangerPtr &cmp, const HostPtr &hostptr, const ThreadPoolPtr &poolptr)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 检查
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                DLOG("已经存在信道%s,不能再次创建", cid.c_str());
                return false;
            }

            auto channel = std::make_shared<Channel>(cid, conn, codec, cmp, hostptr, poolptr);
            _channels.insert(std::make_pair(cid, channel));
            return true;
        }

        void CloseChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }

        ChannelPtr GetChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 检查
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                DLOG("不存在信道%s", cid.c_str());
                return ChannelPtr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, ChannelPtr> _channels;
    };

    using ChannelManagerPtr = std::shared_ptr<ChannelManager>;
}